package abacab.exec;

import java.util.HashMap;
import java.util.Map;

import abacab.ActorSystem;
import abacab.actor.ActorDef;

import com.typesafe.config.Config;
import com.typesafe.config.ConfigException;
import com.typesafe.config.ConfigException.BadValue;

/**
 * A factory for executors
 */
public class MessageExecutorHandler {
	/** The executors */
	private final Map<String, MessageExecutor>	executors;

	/**
	 * Create the factory
	 */
	public MessageExecutorHandler() {
		this.executors = new HashMap<String, MessageExecutor>();
	}

	/**
	 * Stop all executors
	 */
	public void stop() {
		synchronized (this.executors) {
			for (MessageExecutor executor : this.executors.values()) {
				executor.stop();
			}
			this.executors.clear();
		}
	}

	/**
	 * Get the message executor
	 * 
	 * @param system
	 *          the actor system
	 * @param definition
	 *          the actor's definition
	 * @return the dispatcher
	 * @throws ConfigException
	 *           if the configuration is wrong
	 */
	public <M> MessageExecutor getMessageExecutor(ActorSystem system, ActorDef<M> definition) throws ConfigException {
		MessageExecutor executor;
		String name = definition.getExecutorName();
		if (name == null) {
			name = "default";
		}
		synchronized (this.executors) {
			executor = this.executors.get(name);
			if (executor == null) {
				Config executorConfig;
				try {
					executorConfig = system.getConfig().getConfig("abacab.executors." + name);
				}
				catch (ConfigException e) {
					throw new ConfigException.NotResolved("The executor \"" + name + "\" was not found", e);
				}

				// Get the factory
				String factory = executorConfig.getString("factory");
				Class<MessageExecutorFactory> executorClass;
				try {
					executorClass = (Class<MessageExecutorFactory>) Class.forName(factory);
				}
				catch (ClassNotFoundException e) {
					throw new BadValue("abacab.executors." + name + ".factory", "The executor's factory \"" + factory
							+ "\" does not exist", e);
				}
				catch (ClassCastException e) {
					throw new BadValue("abacab.executors." + name + ".factory", "The executor's factory \"" + factory
							+ "\" does not implement \"" + MessageExecutorFactory.class.getName() + "\"");
				}

				// Build the factory
				MessageExecutorFactory executorFactory;
				try {
					executorFactory = executorClass.newInstance();
				}
				catch (Exception e) {
					throw new BadValue("abacab.executors." + name + ".factory", "The executor's factory \"" + factory
							+ "\" can not be built", e);
				}

				executor = executorFactory.buildExecutor(executorConfig);
				this.executors.put(name, executor);
			}
		}
		return executor;
	}
}